#include "EpollServlet.h"
#include "ILogger.h"
#include <sys/epoll.h>
#include <sys/types.h> /* for Socket data types */
#include <sys/socket.h> /* for socket(), connect(), send(), and recv() */
#include <arpa/inet.h>
#include <sys/un.h> /* for sockaddr_un */
#include <netinet/in.h> /* for IP Socket data types */
#include <stdlib.h> /* for exit() */
#include <string.h> /* for memset() */
#include <unistd.h> /* for close() */
#include <errno.h> /* for errno*/

class SendBufferNode {
public:
    std::string content;
    int offset = 0;
};

class EpollWorkSession : public ISession {
public:
    EpollWorkSession(int socket) : socket(socket) {

    }

    int socket = -1;
  
    bool isNeedSend() {
        return sendList.size();
    }
    void reply(const void *data,int len)
    {
        SendBufferNode *node = new SendBufferNode();
        node->content.resize(len);
        memcpy(&node->content[0], data, len);
        std::unique_lock<std::mutex> uniqlock(send_mutex);
        sendList.push_back(node);
    }
    void close()
    {}
    void syncSend() {
        std::unique_lock<std::mutex> uniqlock(send_mutex);
        for (auto i = sendList.begin(); i != sendList.end();) {
            SendBufferNode *node = *i;
            node->offset += send(socket,&node->content[node->offset],node->content.size() - node->offset,0);
            if (node->offset == node->content.size()) {
                i = sendList.erase(i);
                delete node;
            } else {
                break;
            }
        }
    }
    std::list<SendBufferNode *> sendList;
    bool readyInEventPool = false; // 是否已经在发送事件列表里了
    std::mutex send_mutex;
    WorkNode *writeWorkNode = NULL;
    bool isNeedDel = false;

    void crtlEvent(int epollHandler,int type,int events) {
        struct epoll_event ev;
        ev.events = events;
        ev.data.ptr = this;
        if (-1 == epoll_ctl(epollHandler, type, socket, &ev))
        {
            char _buf[100];
            bzero(_buf, sizeof(_buf));
            strerror_r(errno, _buf, sizeof(_buf));
            LogE("error:%s", _buf);
        }
    }
};

IWorkNode::IWorkNode()
{
    
}
void IWorkNode::putTask(int socket)
{
    // 构建session
    std::unique_lock<std::mutex> uniqlock(my_mutex);
    EpollWorkSession *session = new EpollWorkSession(socket);
    sessions.push_back(session);
    v_PrePutTask(session);
}

void IWorkNode::stop()
{
    isRun = false;
    std::unique_lock < std::mutex > uniqlock(my_mutex);
    sessions.clear();
    my_cond.notify_one();
}

void IWorkNode::start()
{
    thr = new std::thread(&IWorkNode::run,this);
}

void IWorkNode::wait()
{
    if (thr) {
        thr->join();
    }
}
IWorkNode::~IWorkNode()
{
    if (thr) {
        delete thr;
        thr = NULL;
    }
}

void IWorkNode::run()
{
    while (isRun)
    {
        std::unique_lock < std::mutex > uniqlock(my_mutex);
        my_cond.wait(uniqlock, [this] {
            if (!isRun)
            {
                LogD("user stopped");
                return true;
            }  
            return this->v_CheckNeedWork();
        });
        v_Action();
        uniqlock.unlock();
    }
}

class WorkNode : public IWorkNode
{
public:
    void v_PrePutTask(EpollWorkSession *session);
    bool v_CheckNeedWork();
    void v_Action();
    void addToWriteEventPool(EpollWorkSession *session);
    void removeFromWriteEventPool(EpollWorkSession *session);

    WorkNode() {
        epollReadHandler = epoll_create(512);
        readEventList.resize(512);

        epollWriteHandler = epoll_create(512);
        writeEventList.resize(512);
    }
public:
    int epollReadHandler;
    std::vector<struct epoll_event> readEventList;
    int epollWriteHandler;
    std::vector<struct epoll_event> writeEventList;
    
    WorkNode *next = NULL;
};


bool WorkNode::v_CheckNeedWork() {
    if (!sessions.empty())
        return true;
    return false;
}

void WorkNode::v_PrePutTask(EpollWorkSession *session)
{
    session->crtlEvent(epollReadHandler,EPOLL_CTL_ADD,EPOLLIN | EPOLLERR | EPOLLPRI | EPOLLHUP);
    if (sessions.size() > readEventList.size()) {
        readEventList.resize(sessions.size());
    }
    if (sessions.size() > readEventList.size()) {
        writeEventList.resize(sessions.size());
    }
    my_cond.notify_one();
}
void WorkNode::addToWriteEventPool(EpollWorkSession *session)
{
    session->crtlEvent(epollWriteHandler,EPOLL_CTL_ADD,EPOLLOUT);
}
void WorkNode::removeFromWriteEventPool(EpollWorkSession *session)
{ 
    session->crtlEvent(epollWriteHandler,EPOLL_CTL_DEL,EPOLLOUT);
}
void WorkNode::v_Action()
{
    // 遍历session
    // 遍历待数据集合
        // 拿到session
        // 处理消息
    int wantToWriteCount = 0;
    for (auto i = sessions.begin(); i != sessions.end();) {
        EpollWorkSession *session = (EpollWorkSession*) *i;
        if (session) {
            if (session->isNeedDel) {
                i = sessions.erase(i);
                delete session;
                continue;
            }
        }
        if (session) {
            if (session->isNeedSend()) {
                if (!session->readyInEventPool) {
                    addToWriteEventPool(session);
                    session->readyInEventPool = true;
                }
                wantToWriteCount++;
            } else {
                 
                // 从event 列表中删除
                removeFromWriteEventPool(session);
                
            }
        }
        ++i;
    }
    // wait 事件
    int retcode = epoll_wait(epollReadHandler, &readEventList[0], (int)sessions.size(), 20);
    if (retcode > 0)
    {
        for(int i = 0; i < retcode; i++)
        {
            EpollWorkSession *session = (EpollWorkSession *)readEventList[i].data.ptr;
            if (readEventList[i].events & (EPOLLERR | EPOLLPRI | EPOLLHUP))
            {
                //套接字出现错误
                LogE("套接字异常错误");
                
                // 标识session 删除
                session->isNeedDel = true;
            }
            else
            {
                if (readEventList[i].events & EPOLLIN)
                {
                    //套接字准备好了读取操作
                    // 连续回去socket 数据 直至取完
                    char buffer[1024] = {0};
                    int size = -1;
                    while ((size = recv(session->socket,buffer,1024,0)) > 0)
                    {
                          server->onMessage(session,buffer,size);
                    }  
                }
            }
            readEventList[i].events=0; 
        }
    }

    // 是否有数据需要发送
    // wait 事件
    if (wantToWriteCount > 0) {
        retcode = epoll_wait(epollWriteHandler, &writeEventList[0], wantToWriteCount, 10);
        if (retcode > 0)
        {
            for(int i = 0; i < retcode; i++)
            {
                EpollWorkSession *session = (EpollWorkSession *)writeEventList[i].data.ptr;
            
                if (writeEventList[i].events & EPOLLOUT)
                {
                    //套接口准备好了写操作
                    session->syncSend();
                    // 删除或更新数据节点
                }
                writeEventList[i].events=0; 
            }
        }
    }
    
}



void EpollServerServlet::on(SERVLET_STATUS type, std::function<int(ISession *session, void *, int)> handler)
{
}
void EpollServerServlet::request(const void *data, int len)
{
}
void EpollServerServlet::start()
{
    listenEpollHandler = epoll_create(1);

    struct sockaddr_in addr;

    if (-1 != serverSocket)
    {
        LogE("服务器可能已经初始化");
        return;
    }

    serverSocket = ::socket(AF_INET, SOCK_STREAM, IPPROTO_TCP);
    if (-1 == serverSocket)
    {
        LogE("创建套接口失败");
        return;
    }

    //设置套接口为可重用状态
    int reuse = 1;
    if (-1 == ::setsockopt(serverSocket, SOL_SOCKET, SO_REUSEADDR, &reuse, sizeof(reuse)))
    {
        LogE("不能设置套接口为可重用状态");
        TEMP_FAILURE_RETRY(::close(serverSocket));
        serverSocket = -1;
        return;
    }

    //设置套接口发送接收缓冲，并且服务器的必须在accept之前设置
    socklen_t window_size = 128 * 1024;
    if (-1 == ::setsockopt(serverSocket, SOL_SOCKET, SO_RCVBUF, &window_size, sizeof(window_size)))
    {
        TEMP_FAILURE_RETRY(::close(serverSocket));
        return;
    }
    if (-1 == ::setsockopt(serverSocket, SOL_SOCKET, SO_SNDBUF, &window_size, sizeof(window_size)))
    {
        TEMP_FAILURE_RETRY(::close(serverSocket));
        return;
    }

    bzero(&addr, sizeof(addr));
    addr.sin_family = AF_INET;
    addr.sin_addr.s_addr = htonl(INADDR_ANY);
    addr.sin_port = htons(port);

    int retcode = ::bind(serverSocket, (struct sockaddr *)&addr, sizeof(addr));
    if (-1 == retcode)
    {
        LogE("不能绑定服务器端口");
        TEMP_FAILURE_RETRY(::close(serverSocket));
        serverSocket = -1;
        return;
    }

    retcode = ::listen(serverSocket, 2000);
    if (-1 == retcode)
    {
        LogE("监听套接口失败");
        TEMP_FAILURE_RETRY(::close(serverSocket));
        serverSocket = -1;
        return;
    }

    struct epoll_event ev;
    ev.events = EPOLLIN;
    ev.data.ptr = NULL;
    if (!epoll_ctl(listenEpollHandler, EPOLL_CTL_ADD, serverSocket, &ev))
    {
        LogE("绑定Epoll失败");
        return;
    }

    LogI("初始化 %u 成功", port);

    listenThread = new std::thread(&EpollServerServlet::acceptThreadLogic,this);
    return;
}
void EpollServerServlet::wait()
{
    if (listenThread) {
        listenThread->join();
    }
}
void EpollServerServlet::stop()
{
    isRun = false;
}
void EpollServerServlet::onLoad(INode *node)
{
    INode * child = node->childNode("listen");
    if (child) {
        port = child->get<int>("port");
    }

    INode * threads = node->childNode("threads");
    if (threads) {
        int count = threads->get<int>("count");
        for (int i = 0; i < count; ++i) {
            WorkNode *workNode = new WorkNode();
            workNode->next = workList;
            workNode->server = this;
            workList = workNode;
            workNode->start();
        }
        currLocation = workList;
    }
}
void EpollServerServlet::onUnload()
{
    WorkNode *temp = workList;
    while (temp) {
        temp->stop();
        temp = temp->next;
    }
}
void EpollServerServlet::onMessage(ISession *session, void *data, int size)
{
}
void EpollServerServlet::onState(SERVLET_STATUS status, ISession *session, int code)
{
}
EpollServerServlet::~EpollServerServlet()
{
    if (listenThread) {
        delete listenThread;
        listenThread = NULL;
    }
}

void EpollServerServlet::putToWorkThread(int socket)
{
    // 读线程池
    if (currLocation)
    {
        currLocation->balance++;
        currLocation->putTask(socket);
        if (currLocation->next)
        {
            currLocation = currLocation->next;
        }
        else
        {
            currLocation = workList;
        }
    }
   
}
void EpollServerServlet::acceptThreadLogic()
{
    while (isRun)
    {
        struct sockaddr_in addr;
        socklen_t len = sizeof(struct sockaddr_in);
        bzero(&addr, sizeof(struct sockaddr_in));

        struct epoll_event ev;
        int rc = epoll_wait(listenEpollHandler, &ev, 1, 100);
        if (1 == rc && (ev.events & EPOLLIN))//准备好接受
        {
            int socket = TEMP_FAILURE_RETRY(::accept(serverSocket, (struct sockaddr *)&addr, &len));
            putToWorkThread(socket);
        }
    }
}



void EpollClientServlet::on(SERVLET_STATUS type,std::function<int(ISession *session,void*,int)> handler)
{
    msgidHandlers[type].push_back(handler);
}
void EpollClientServlet::request(const void *data,int len)
{
    int i = send(conn_fd, data, len, 0);
    LogI("EpollClientServlet::request want:%d real:%d\n",len,i);
    // 放入缓存
}

void EpollClientServlet::recvthreadLogic(int socket) {

    struct timeval tv_out;
    tv_out.tv_sec = 3;
    tv_out.tv_usec = 0;
    setsockopt(socket, SOL_SOCKET, SO_RCVTIMEO, &tv_out, sizeof(tv_out));
    EpollWorkSession clientSession(socket);
    onState(OPEN,&clientSession,0);

    int clientEpollHandler = epoll_create(1);

    // 注册读入事件

    clientSession.crtlEvent(clientEpollHandler,EPOLL_CTL_ADD, EPOLLIN | EPOLLERR | EPOLLPRI | EPOLLHUP);
    std::vector<struct epoll_event> eventList;
    eventList.resize(1);
    while (isRun) {
        if (clientSession.isNeedSend()) {
            // 加入write事件
            clientSession.crtlEvent(clientEpollHandler,EPOLL_CTL_ADD,EPOLLOUT);
        }
        else {
            //删除write事件
            clientSession.crtlEvent(clientEpollHandler,EPOLL_CTL_DEL,EPOLLOUT);
        }

        int retcode = epoll_wait(clientEpollHandler, &eventList[0], 1, 20);
        if (retcode > 0)
        {
            for(int i = 0; i < retcode; i++)
            {
                EpollWorkSession *session = (EpollWorkSession *)eventList[i].data.ptr;
                if (eventList[i].events & (EPOLLERR | EPOLLPRI | EPOLLHUP))
                {
                    //套接字出现错误
                    LogE("套接字异常错误");
                    
                    // 标识session 删除
                    session->isNeedDel = true;
                }
                else
                {
                    if (eventList[i].events & EPOLLIN)
                    {
                        //套接字准备好了读取操作
                        // 连续回去socket 数据 直至取完
                        char buffer[1024] = {0};
                        int size = -1;
                        while ((size = recv(session->socket,buffer,1024,0)) > 0)
                        {
                            onMessage(session,buffer,size);
                        }  
                    }
                    if (eventList[i].events & EPOLLOUT)
                    {
                        session->syncSend();
                    }
                }
                eventList[i].events=0; 
            }
        }
        
    }
    LogI("EpollClientServlet::recvthreadLogic end\n");
}
void EpollClientServlet::threadLogic()
{
    struct sockaddr_in  servaddr;

    if( (conn_fd = socket(AF_INET, SOCK_STREAM, 0)) < 0){
        LogI("create socket error: %s(errno: %d)", strerror(errno),errno);
        return;
    }

    memset(&servaddr, 0, sizeof(servaddr));
    servaddr.sin_family = AF_INET;
    servaddr.sin_port = htons(port);
    if( inet_pton(AF_INET,ip.c_str(), &servaddr.sin_addr) <= 0){
        LogI("inet_pton error for %s:%d",ip.c_str(),port);
        return;
    }

    int retryCount = 0;
    while(connect(conn_fd, (struct sockaddr *)&servaddr, sizeof(servaddr)) == -1 && retryCount++ < 3)
    {
        LogI("connect error retryCount:%d",retryCount);
        std::this_thread::sleep_for(std::chrono::milliseconds(1500));
    }
    LogI("connect successfully!");
    auto client = std::thread(&EpollClientServlet::recvthreadLogic,this,conn_fd);
    client.join();
    close(conn_fd);
    return;
}
void EpollClientServlet::start()
{
    // create a thread handle things in a thread
    _thread = new std::thread(&EpollClientServlet::threadLogic,this);
}
void EpollClientServlet::wait()
{
    ((std::thread*)_thread)->join();
    
}
void EpollClientServlet::stop()
{
    isRun = false;
}
void EpollClientServlet::onLoad(INode *node)
{
    INode *child = node->childNode("connection");
    if (child) {
        port = child->get<int>("port");
        ip = child->getStr("ip");
    }
    start();
}
void EpollClientServlet::onMessage(ISession *session,void* data,int size)
{
    auto iter = msgidHandlers.find(STREAM);
    if (iter != msgidHandlers.end()) {
        for (auto i = iter->second.begin(); i != iter->second.end();++i)
           (*i)(session,data,size);
    }
}
void EpollClientServlet::onState(SERVLET_STATUS status,ISession *session,int code)
{
    auto iter = msgidHandlers.find(status);
    if (iter != msgidHandlers.end()) {
         for (auto i = iter->second.begin(); i != iter->second.end();++i)
           (*i)(session,NULL,code);
    }
}

void EpollClientServlet::onUnload()
{
    stop();
    wait();
}
EpollClientServlet::~EpollClientServlet()
{
     delete (std::thread*)_thread; 
}
